

package com.example.example01;

import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.net.URISyntaxException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
import org.apache.hadoop.io.IOUtils;

/**
 * hadoop client工具类
 * 
 * 用于快速组织hadooop操作
 * 
 */
public class HadoopClient {
  private String hdfsUrl;
  private String hdfsUsr;
  /**
   * 获取hadoop配置信息
   * @return Configuration
   */
  private Configuration getConfiguration(){
    Configuration configuration = new Configuration();
    return configuration;
  }
  public String getHdfsUrl() {
    return hdfsUrl;
  }
  public void setHdfsUrl(String hdfsUrl) {
    this.hdfsUrl = hdfsUrl;
  }
  public String getHdfsUsr() {
    return hdfsUsr;
  }
  public void setHdfsUsr(String hdfsUsr) {
    this.hdfsUsr = hdfsUsr;
  }
  /**
   * hdfs的filesystem
   *
   * 需要自行关闭
   *
   * @return
   * @throws IOException
   * @throws InterruptedException
   * @throws URISyntaxException
   */
  private FileSystem getHdfsFileSystem() throws IOException, InterruptedException, URISyntaxException{
    String hdfsUrl = this.getHdfsUrl();
    String hdfsUsr = this.getHdfsUsr();
    if(hdfsUrl == null || hdfsUrl == "")
      throw new IllegalArgumentException("hdfsUrl不可为空");
    if(hdfsUsr == null || hdfsUsr == "")
      throw new IllegalArgumentException("hdfsUsr不可为空");
    URI uri = new URI(hdfsUrl);
    Configuration config = this.getConfiguration();
    FileSystem fs = FileSystem.get(uri,config, hdfsUsr);
    return fs;
  }

  /**
   * 获取文件列表（包括文件夹）
   *
   * @param path
   * @param isRecursive
   * @return
   * @throws URISyntaxException
   * @throws InterruptedException
   * @throws IOException
   */
  public RemoteIterator<LocatedFileStatus> getList(String path,boolean isRecursive) throws IOException, InterruptedException, URISyntaxException{
    if(path == null || path == "")
      return null;
    Path pth = new Path(path);
    // 返回之后自动关闭问题怎么解决？
    // try(FileSystem fs = this.getHdfsFileSystem()){
      FileSystem fs = this.getHdfsFileSystem();
      RemoteIterator<LocatedFileStatus> fileStatusIter = fs.listFiles(pth, isRecursive);
      return fileStatusIter;
    // }catch(Exception ex){
    //   throw ex;
    // }
  }

  /**
   * 上传文件
   * @param sourcePath
   * @param saveToPath
   * @throws URISyntaxException
   * @throws InterruptedException
   * @throws IOException
   */
  public void uploadFile(String sourcePath,String saveToPath) throws IOException, InterruptedException, URISyntaxException{
    if(sourcePath == null || sourcePath == "")
      return;
    Path sorPath = new Path(sourcePath);
    Path savPath = new Path(saveToPath);
    try(FileSystem fs = this.getHdfsFileSystem()){
      fs.copyFromLocalFile(sorPath,savPath);
    }catch(Exception ex){
      throw ex;
    }
  }
  /**
   * 通过流的方式创建、上传文件
   * @param sourceStream
   * @param saveToPath
   * @throws IOException
   * @throws InterruptedException
   * @throws URISyntaxException
   */
  public void uploadFile(InputStream sourceStream,String saveToPath) throws IOException, InterruptedException, URISyntaxException{
    if(sourceStream == null)
      return;
    Path savPath = new Path(saveToPath);
    Configuration config = this.getConfiguration();
    try(FileSystem fs = this.getHdfsFileSystem()){
      FSDataOutputStream outputStream = fs.create(savPath);
      IOUtils.copyBytes(sourceStream, outputStream,config);
      IOUtils.closeStream(sourceStream);
      IOUtils.closeStream(outputStream);
    }catch(Exception ex){
      throw ex;
    }
  }

  /**
   * 下载文件
   * @param sourcePath
   * @param saveToPath
   * @throws Exception
   */
  public void downloadFile(String sourcePath,String saveToPath) throws IOException,InterruptedException,URISyntaxException,IllegalArgumentException {
    if(sourcePath == null || sourcePath == "")
      return;
    Path sorPath = new Path(sourcePath);
    Path savPath = new Path(saveToPath);
    try(FileSystem fs = this.getHdfsFileSystem()){
      fs.copyToLocalFile(false,sorPath,savPath,true);
    }catch(Exception ex){
      throw ex;
    }
  }
  /**
   * 创建文件夹
   * 
   * @param dirPath
   * @throws IOException
   * @throws InterruptedException
   * @throws URISyntaxException
   */
  public void makeDir(String dirPath) throws IOException, InterruptedException, URISyntaxException{
    try(FileSystem fileSystem = this.getHdfsFileSystem()){
      Path pth = new Path(dirPath);
      fileSystem.mkdirs(pth);
    }catch(Exception ex){
      throw ex;
    }
  }
  /**
   * 删除文件/文件夹
   * 
   * @param sourcePath
   * @throws IOException
   * @throws URISyntaxException
   * @throws InterruptedException
   * @throws IllegalArgumentException
   */
  public void delete(String sourcePath) throws IOException,URISyntaxException,InterruptedException,IllegalArgumentException{
    Path sorPath = new Path(sourcePath);
    try(FileSystem fileSystem = this.getHdfsFileSystem()){
      fileSystem.delete(sorPath,true);
    }catch(Exception ex){
      throw ex;
    }
  }
}
